package distributedbank;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;

import distributedbank.server.messages.Acknowledgment;
import distributedbank.server.messages.OracleKill;
import distributedbank.server.messages.OracleNotification;
import distributedbank.server.messages.Request;
import distributedbank.server.messages.SyncRequest;
import distributedbank.server.messages.SyncRequestAck;
import distributedbank.server.messages.Transaction;

public class Wrapper {

    // each branch server socket will listen on: basePort + branchId * 10 +
    // backupId
    // each gui socket will listen on: basePort + branchId * 2
    public static final int basePort = 10011;

    private static final String topologyFile = "topology_file.txt";

    private final ServerSocket serverSocket;

    public final int localPort;

    public List<Set<Integer>> servers;

    // incoming messages that haven't been consumed, yet
    public final List<Serializable> messageQueue = new LinkedList<Serializable>();

    // specifies which jvm's can communicate with each other
    private final Map<Integer, Set<Integer>> topology;

    /**
     * Creates a wrapper instance for the given branchId. The wrapper represents
     * a BranchGui if isGui is true. Represents a BankBranch server, otherwise.
     * 
     * Every branch may have at most 9 backup servers.
     * 
     */
    public Wrapper(int branchId, int backupId, boolean isGui)
            throws IOException {
        if (isGui) {
            // backupId is irrelevant for GUI processes
            this.localPort = basePort + branchId * 10;
        } else {
            this.localPort = basePort + branchId * 10 + backupId;
        }
        this.serverSocket = new ServerSocket(this.localPort);
        this.topology = Wrapper.readTopologyFile();
    }

    public Wrapper(int branchId, int backupId, boolean isGui,
            List<Set<Integer>> servers) throws IOException {
        this(branchId, backupId, isGui);
        this.servers = servers;
    }

    /**
     * Call this every time oracle pushes a list of servers
     */
    public void setAvailableServers(List<Set<Integer>> servers) {
        this.servers = servers;
    }

    /**
     * Reads the contents of the topology file, and returns a hash-set that
     * specifies which JVM's may communicate to each other.
     */
    private static Map<Integer, Set<Integer>> readTopologyFile()
            throws IOException {
        FileReader file = new FileReader(Wrapper.topologyFile);
        BufferedReader stream = new BufferedReader(file);

        Map<Integer, Set<Integer>> result = new HashMap<Integer, Set<Integer>>();

        // parse each line of the file...
        try {
            while (true) {
                String line = stream.readLine();
                if (line == null) {
                    break;
                } else if (line.length() == 0) {
                    continue;
                }
                String[] contents = line.split(" ");

                // check for errors
                if (contents.length != 2) {
                    throw new IOException("Error: topology file is malformed!");
                }

                int[] ports = new int[2];
                for (int i : new int[] { 0, 1 }) {
                    String string = contents[i].toLowerCase();

                    // check for more errors
                    if (string.startsWith("server") && string.length() == 8) {
                        int branchId = Integer.valueOf(string.substring(6));

                        // servers from a given branch can call communicate with
                        // the same servers
                        ports[i] = Wrapper.basePort + branchId * 10;
                    } else {
                        throw new IOException(
                                "Error: topology file is malformed!"
                                        + "Every entry must be of the"
                                        + "form 'client01' or 'server02.3'");
                    }
                }

                // insert the line into our result object
                for (int j = 0; j < 10; j++) {
                    if (!result.containsKey(ports[0] + j)) {
                        result.put(ports[0] + j, new HashSet<Integer>());
                    }
                    for (int k = 0; k < 10; k++) {
                        result.get(ports[0] + j).add(ports[1] + k);
                    }
                }
            }
            return result;
        } finally {
            stream.close();
            file.close();
        }
    }

    /**
     * This will send a request, and will block until an appropriate
     * Acknowledgment is received. If the oracle determines that the server
     * we're sending to is down, the message is re-sent to one of the backup
     * servers.
     * 
     * Returns null if the destination is unreachable.
     */
    public Acknowledgment sendReliable(Transaction request) throws IOException,
            ClassNotFoundException {

        int backupId = -1;
        while (true) {
            Set<Integer> availableServers = this.servers
                    .get(request.branchId - 1);

            if (availableServers.size() == 0) {
                System.out.println("No available servers for branch "
                        + request.branchId);
                return null;
            }

            if (!availableServers.contains(backupId)) {
                backupId = Collections.min(availableServers);
                if (!this.sendMessageToPort(request, basePort + 10
                        * request.branchId + backupId)) {
                    System.out
                            .println("Cannot send messages to port of branch "
                                    + request.branchId);
                    return null;
                }
            }

            receiveLowLevel(); // the incoming message will be appended to the
                               // message queue
            Serializable incomingMessage = this.messageQueue
                    .get(this.messageQueue.size() - 1);

            if (incomingMessage instanceof Acknowledgment) {
                this.messageQueue.remove(this.messageQueue.size() - 1);

                Acknowledgment ack = (Acknowledgment) incomingMessage;

                if (ack.fromPort == basePort + 10 * request.branchId + backupId
                        && ack.transactionId == request.transactionId) {
                    return ack;
                } else {
                    // ignore all other acknowledgments... they must be junk
                }
            } else if (incomingMessage instanceof OracleNotification) {
                // the low-level receive function will already have dealt with
                // this type of message
                this.messageQueue.remove(this.messageQueue.size() - 1);
            } else {
                // don't do anything; wait for appropriate acknowledgment
            }
        }
    }

    public void sendSyncRequests(Transaction transaction) throws IOException,
            ClassNotFoundException {
        Set<Integer> availableServers = new TreeSet<Integer>();

        availableServers.addAll(this.servers.get(transaction.branchId - 1));
        availableServers.remove(this.localPort - basePort - 10
                * transaction.branchId);
        SyncRequest request = new SyncRequest(transaction.branchId,
                this.localPort, transaction);

        for (Integer server : availableServers) {
            this.sendMessageToPort(request, basePort + 10
                    * transaction.branchId + server);
        }

        System.out.println(availableServers);

        // get messages from all servers that are still up
        while (availableServers.size() > 0) {
            availableServers.retainAll(this.servers
                    .get(transaction.branchId - 1));

            receiveLowLevel();
            Serializable incomingMessage = this.messageQueue
                    .get(this.messageQueue.size() - 1);

            if (incomingMessage instanceof SyncRequestAck) {
                this.messageQueue.remove(this.messageQueue.size() - 1);

                SyncRequestAck ack = (SyncRequestAck) incomingMessage;

                int backupId = ack.fromPort - basePort - 10 * request.branchId;
                System.out.println(backupId);
                if (availableServers.contains(backupId)
                        && transaction.transactionId == ack.transactionId) {
                    availableServers.remove(backupId);
                } else {
                    // ignore all other acknowledgments... they must be junk
                }
            } else if (incomingMessage instanceof OracleNotification) {
                // the low-level receive function will already have dealt with
                // this type of message
                this.messageQueue.remove(this.messageQueue.size() - 1);
            } else {
                // don't do anything; wait for appropriate acknowledgment
            }
        }
    }

    /**
     * Retrieves a serializable message from the incoming socket and adds it to
     * the message queue.
     */
    public void receiveLowLevel() throws IOException, ClassNotFoundException {
        Socket clientSocket = this.serverSocket.accept();
        ObjectInputStream in = new ObjectInputStream(
                clientSocket.getInputStream());

        Serializable message;
        try {
            message = (Serializable) in.readObject();
        } finally {
            in.close();
            clientSocket.close();
        }

        this.messageQueue.add(message);

        if (message instanceof OracleNotification) {
            this.servers = ((OracleNotification) message).servers;
        } else if (message instanceof OracleKill) {
            System.out.println("Killed server at port " + this.localPort);
            System.exit(0);
        }
    }

    /**
     * This function is called by all servers/guis when they start up. It waits
     * until the first notification from the oracle is received.
     */
    public void waitForInitialOracleNotification() throws IOException,
            ClassNotFoundException {
        while (true) {
            receiveLowLevel();

            Serializable message = this.messageQueue.get(this.messageQueue
                    .size() - 1);
            if (message instanceof OracleNotification) {
                this.messageQueue.remove(this.messageQueue.size() - 1);
                return;
            }
        }
    }

    /**
     * Retrieves a message from the incoming socket. The returned message will
     * always be a "Request" object.
     */
    public Serializable receive() throws IOException, ClassNotFoundException {
        while (true) {
            if (this.messageQueue.size() == 0) {
                receiveLowLevel();
            }

            Serializable message = this.messageQueue.remove(0);
            if (message instanceof Request) {
                return message;
            }
        }
    }

    /**
     * Send a message to a gui client. Returns true on success, false on
     * failure.
     */
    public boolean sendGuiMessage(Serializable message, int branchId)
            throws IOException {
        // no need to change this because there is no pb for client
        return this
                .sendMessageToPort(message, Wrapper.basePort + 10 * branchId);
    }

    /**
     * Send a response to primary server
     */
    // public boolean send(SynAckRequest message) throws IOException {
    // // we may need to reconfigure the port number...
    // return this.sendMessageToPort(message, Wrapper.basePort + 2
    // * getPrimaryServerByBranchId(message.branchId));
    // }

    /**
     * Returns a set of all ports reachable by this wrapper.
     */
    public Set<Integer> whoNeighbors() {
        if (this.topology.containsKey(this.localPort)) {
            return this.topology.get(this.localPort);
        } else {
            return new HashSet<Integer>();
        }
    }

    /**
     * Returns the set of branchIds for all BankBranch servers that can be
     * reached from this JVM.
     */
    public Set<Integer> whoNeighborBranches() {
        Set<Integer> result = new HashSet<Integer>();
        for (int i : this.whoNeighbors()) {
            if ((i - Wrapper.basePort) % 2 == 0) {
                // this is a server port
                result.add((i - Wrapper.basePort) / 2);
            }
        }
        return result;
    }

    /**
     * Sends an acknowledgment message to the server at the given port.
     */
    public boolean sendAcknowledgment(Acknowledgment message, int port)
            throws IOException {
        return this.sendMessageToPort(message, port);
    }

    /**
     * Attempts to send a Message to the VM with the given id.
     */
    private boolean sendMessageToPort(Serializable message, int port) {
        if (!this.whoNeighbors().contains(port)) {
            // throw new IOException(
            // "Topology file does not permit communication with requested JVM.")
            // ;
            return false;
        }

        try {
            Socket socket = new Socket("localhost", port);
            ObjectOutputStream out = new ObjectOutputStream(
                    socket.getOutputStream());
            try {
                out.writeObject(message);
            } finally {
                out.close();
                socket.close();
            }
        } catch (IOException e) {
            System.out.println("Failed to send message to port: " + port);
        }

        return true;
    }

    /**
     * The oracle can communicate with everybody. No need for the config file!
     */
    public static void sendOracleMessage(Serializable message, int port)
            throws IOException {
        Socket socket = new Socket("localhost", port);
        ObjectOutputStream out = new ObjectOutputStream(
                socket.getOutputStream());
        try {
            out.writeObject(message);
        } finally {
            out.close();
            socket.close();
        }
    }

}
